// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_LOCATION_CACHE_HPP_
#define _XBASE_LOCATION_CACHE_HPP_

#include <map>
#include <list>
#include <string>

#include <boost/function.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/intrusive_ptr.hpp>
#include <boost/system/error_code.hpp>

#include "protocol.hpp"
#include "session_manager.hpp"
#include "root_meta_cache.hpp"
#include "io_service_pool.hpp"
#include "async_module_base.hpp"
#include "meta_tablet_scanner.hpp"

namespace xbase
{
	struct location_cache_key
	{
		std::string table_name;
		std::string start_row;

		std::string to_string();

		static std::string to_string(const location_cache_key &key);
	};

	struct location_cache_value
	{
		uint64_t meta_got_time;
		protocol::tablet_info_ptr tablet_info_p;
		protocol::server_address_ptr server_addr_p;

		std::string to_string();

		static std::string to_string(const location_cache_value &value);
	};

	inline bool operator<(const location_cache_key &left_hand_side, const location_cache_key &right_hand_side)
	{
		if (left_hand_side.table_name != right_hand_side.table_name)
			return left_hand_side.table_name < right_hand_side.table_name;

		return (left_hand_side.start_row < right_hand_side.start_row);
	}

	class location_cache
		: public async_module_base<io_service_pool::xbase_level, io_service_pool::xbase_level>
	{
	public:
		typedef async_module_base<io_service_pool::xbase_level, io_service_pool::xbase_level> base_type;
		typedef location_cache this_type;
		typedef boost::intrusive_ptr<this_type> ptr_type;

		typedef meta_tablet_scanner::ptr_type meta_tablet_scanner_ptr;
		typedef session_manager::endpoint_type endpoint_type;

		typedef boost::system::error_code error_code;

		typedef std::map<location_cache_key, location_cache_value> meta_data_map;

		typedef protocol::tablet_info_ptr tablet_info_ptr;
		typedef protocol::server_address_ptr server_address_ptr;
		typedef boost::function<void(location_cache_value, const error_code&)> handler_type;

		typedef std::pair<std::string, handler_type> new_table_scan_waiting_item;
		typedef std::list<new_table_scan_waiting_item> new_table_scan_waiting_list;
		typedef std::map<std::string, new_table_scan_waiting_list> new_table_scan_waiting_map;

	public:
		struct request_item
		{
			handler_type m_handler;
			std::string m_table_name;
			std::string m_request_row;
			std::string m_start_row;
			std::string m_end_row;
			unsigned m_range_expand_times;
			uint64_t m_old_meta_got_time;
			meta_tablet_scanner_ptr m_meta_tablet_scanner_ptr;

			request_item(handler_type handler, std::string table_name, std::string request_row,
				std::string start_row, std::string end_row, uint64_t old_meta_got_time)
				: m_handler(handler), m_table_name(table_name), m_request_row(request_row),
				m_start_row(start_row), m_end_row(end_row), m_range_expand_times(0),
				m_old_meta_got_time(old_meta_got_time)
			{}

			bool is_whole_table_scan() const;
		};
		typedef boost::shared_ptr<request_item> request_item_ptr;
		request_item_ptr null_request_item_ptr;

		typedef std::list<request_item_ptr> request_list;

	public:
		location_cache(io_service_pool &pool,
			root_meta_cache &rmc,
			session_manager &sm)
			: base_type(pool),
			m_io_service_pool(pool),
			m_max_update_steps(3),
			m_root_meta_valid(false),
			m_root_meta_cache(rmc),
			m_session_manager(sm)
		{
		}

		void async_get_meta(handler_type handler,
			std::string table_name,
			std::string request_row,
			bool hard,
			uint64_t old_meta_got_time)
		{
			this->get_lower_level_service().dispatch(
				boost::bind(
				&this_type::on_async_get_meta, this_ptr(), handler,
				table_name, request_row, hard, old_meta_got_time
				)
				);
		}

	private:
		void on_async_get_meta(handler_type handler, std::string table_name,
			std::string request_row, bool hard, uint64_t old_meta_got_time);

		void on_got_root_meta(const error_code& ec);

		void on_got_next_row(meta_row_result::ptr_type meta_row_ptr, const error_code &ec);

		void process_new_request();

		void expand_range(meta_data_map::iterator iter, request_item_ptr item_ptr);

		meta_data_map::iterator update(meta_row_result::ptr_type meta_row_ptr);

		bool lookup(std::string &table_name, std::string &request_row, meta_data_map::iterator &iter);

		void do_root_meta_query();

		void do_new_meta_scan(request_item_ptr item_ptr);

		void do_next_meta_row_scan(request_item_ptr item_ptr);

		void make_new_request(handler_type handler, std::string &table_name, std::string &request_row,
			std::string &scan_start_row, std::string &scan_end_row, uint64_t old_meta_got_time);

		void dispatch(std::string &request_row, handler_type handler, location_cache_value meta, error_code ec);

	protected:
		// 缓冲模块不需要用ptr_type来持有
		this_type *this_ptr() {
			return this;
		}

	private:
		io_service_pool &m_io_service_pool;
		meta_data_map m_meta_data_map;

		const unsigned m_max_update_steps;

		bool m_root_meta_valid;
		request_list m_meta_request_list;
		root_meta_cache &m_root_meta_cache;

		session_manager &m_session_manager;

		new_table_scan_waiting_map m_new_table_scan_waiting_map;
	};
	
	typedef location_cache::ptr_type location_cache_ptr;
}

#endif
